//
// Created by 11510 on 2024/7/9.
//

#ifndef REACTOR_V2_EVENTLOOP_H
#define REACTOR_V2_EVENTLOOP_H

#include "Acceptor.h"
#include "TcpConnection.h"
#include <sys/epoll.h>
#include <vector>
#include <map>
#include <memory>
#include <functional>
#include <mutex>
#include <sys/eventfd.h>

class EventLoop {
public:
    using TcpConnectionPtr = std::shared_ptr<TcpConnection>;
    using TcpConnectionCallback = std::function<void(const TcpConnectionPtr &)>;
    using Functor = std::function<void()>;
private:
    int _epfd;  /* epoll_create创建出的文件描述符 */
    std::vector<struct epoll_event> _evtList;   /* 存放满足条件的文件描述符数据结构 */
    bool _isLooping;    /* 表示循环是否运行*/
    Acceptor &_acceptor;    /* 连接器Acceptor的引用 */
    std::map<int, TcpConnectionPtr> _conns;   /* 存放文件描述符与TcpConnection键值对 */

    TcpConnectionCallback _onNewConnectionCallback;
    TcpConnectionCallback _onMessageCallback;
    TcpConnectionCallback _onCloseCallback;

    /**
     * eventfd
     */
    int _evtfd;     /* 线程通信文件描述符 */
    std::vector<Functor> _pendings;     /* 待执行的任务 */
    std::mutex _mutex;      /* 互斥访问vector */


public:
    EventLoop(Acceptor &acceptor) :
            _epfd(this->createEpollFd()),
            _evtList(1024),
            _isLooping(false),
            _acceptor(acceptor),
            _conns(),
            _evtfd(this->createEventFd()),
            _mutex() {
        int listenfd = _acceptor.getFileDescriptor();
        addEpollReadFd(listenfd);

        /* 添加监听线程通信文件描述符 */
        addEpollReadFd(_evtfd);
    }

    ~EventLoop() {
        close(_epfd);
        close(_evtfd);
    }

    void loop() {
        _isLooping = true;
        while (_isLooping) {
            waitEpollFd();
        }

    }

    void unloop() {
        _isLooping = false;
    }

    void setOnNewConnectionCallback(TcpConnectionCallback &&onNewConnectionCallback) {
        _onNewConnectionCallback = std::move(onNewConnectionCallback);
    }

    void setOnMessageCallback(TcpConnectionCallback &&onMessageCallback) {
        _onMessageCallback = std::move(onMessageCallback);
    }

    void setOnCloseCallback(TcpConnectionCallback &&onCloseCallback) {
        _onCloseCallback = std::move(onCloseCallback);
    }

    int createEventFd() {
        int ret = eventfd(0, 0);
        if (ret < 0) {
            perror("eventfd");
            return ret;
        }
        return ret;
    }

    void handleRead() {
        uint64_t one;
        ssize_t ret = read(_evtfd, &one, sizeof one);
        if (ret != sizeof one) {
            perror("read");
            return;
        }
    }

    void wakeup() {
        uint64_t one = 1;
        ssize_t ret = write(_evtfd, &one, sizeof one);
        if (ret != sizeof one) {
            perror("write");
            return;
        }
    }

    /**
     * 遍历vector将所有任务执行
     */
    void doPendingFunctors() {

        std::vector<Functor> tmp;

        {
            std::lock_guard<std::mutex> lockGuard(_mutex);
            tmp.swap(_pendings);
        }

        for (auto &cb: tmp) {
            cb();
        }
    }

    /**
     * Functor的参数this, msg已经在bind中传完了，所以形参列表为空
     * @param cb
     */
    void runInLoop(Functor &&cb) {
        {
            /* 控制锁的粒度 */
            std::lock_guard<std::mutex> lockGuard(_mutex);
            _pendings.push_back(std::move(cb));
        }
        wakeup();
    }

private:
    int createEpollFd() {
        return epoll_create1(0);
    }

    void addEpollReadFd(int fd) {
        struct epoll_event evt;
        evt.data.fd = fd;
        evt.events = EPOLLIN;

        int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &evt);
        if (ret < 0) {
            perror("addEpollReadFd");
            return;
        }
    }

    void removeEpollReadFd(int fd) {
        struct epoll_event evt;
        evt.data.fd = fd;
        evt.events = EPOLLIN;

        int ret = epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, &evt);
        if (ret < 0) {
            perror("removeEpollReadFd");
            return;
        }
    }

    void waitEpollFd() {
        int nready;
        do {
//            获得首地址&*
            nready = ::epoll_wait(_epfd, &*_evtList.begin(), _evtList.size(), 3000);
        } while (-1 == nready && errno == EINTR);

        if (-1 == nready) {
            perror("-1 === nready");
            return;
        } else if (0 == nready) {
            printf(">>epoll_wait time out!!!\n");

        } else {
//            BUG ? ? ?
            /**
             * vector有可能越界，需要手动扩容
             * nready有可能是-1（有这个可能？）
             */

            if ((int) _evtList.size() == nready) {
                _evtList.resize(2 * nready);
            }

            for (size_t idx = 0; idx < nready; idx++) {
                int fd = _evtList[idx].data.fd;
                int listenfd = _acceptor.getFileDescriptor();

                if (fd == listenfd) {
                    /* 新的连接 */
                    handleNewConnection();
                } else if (fd == _evtfd) {
                    handleRead();
                    /**
                     * 此处任务是线程池处理之后的msg?
                     */
                    doPendingFunctors();/* 执行所有"任务" */
                } else {
                    /* 有连接发信息 */
                    handleNewMessage(fd);
                }
            }
        }

    }

    void handleNewConnection() {
        int connfd = _acceptor.accept();
        if (connfd < 0) {
            perror("handleNewConnection");
            return;
        }
        /* 监听connfd */
        addEpollReadFd(connfd);

        /* 创建TcpConnection对象并插入 */
        TcpConnectionPtr con(new TcpConnection(connfd, this));
        con->setOnNewConnectionCallback(_onNewConnectionCallback);  /* 连接建立 */
        con->setOnMessageCallback(_onMessageCallback);  /* 消息到达 */
        con->setOnCloseCallback(_onCloseCallback);  /* 连接断开 */

//        _conns.insert(std::make_pair(connfd, con));   /*方式一*/
        _conns[connfd] = con;   /* 方式二 */

        con->handleOnNewConnectionCallback();
    }

    void handleNewMessage(int fd) {
        std::map<int, TcpConnectionPtr>::iterator it = _conns.find(fd);

        if (it != _conns.end()) {
            if (it->second->isClosed()) {
                printf("亖了一个\n");
                it->second->handleOnCloseCallback();
                removeEpollReadFd(fd);
                /**
                 * 智能指针引用为0，自动调用析构，既close()
                 */
                _conns.erase(it);

            } else {
                it->second->handleOnMessageCallback();
            }
        } else {
            printf("连接不存在\n");
            return;
        }
    }
};


#endif //REACTOR_V2_EVENTLOOP_H
